package com.example.test.stream8;

import com.example.constant.KafkaTopic;
import com.example.entity.stream1.*;
import com.example.entity.stream8.*;
import io.confluent.kafka.serializers.AbstractKafkaAvroSerDeConfig;
import io.confluent.kafka.serializers.KafkaAvroSerializer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class KafkaProducerABCDEFGHIJExample {


    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.8.80:9092,192.168.8.80:9093,192.168.8.80:9094");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class.getName());
        props.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://192.168.8.80:8081");

        Producer<String, A8> producerA = new KafkaProducer<>(props);
        Producer<String, B8> producerB = new KafkaProducer<>(props);
        Producer<String, C8> producerC = new KafkaProducer<>(props);
        Producer<String, D8> producerD = new KafkaProducer<>(props);
        Producer<String, E8> producerE = new KafkaProducer<>(props);
        Producer<String, F8> producerF = new KafkaProducer<>(props);
        Producer<String, G8> producerG = new KafkaProducer<>(props);
        Producer<String, H8> producerH = new KafkaProducer<>(props);
        Producer<String, I8> producerI = new KafkaProducer<>(props);
        Producer<String, J8> producerJ = new KafkaProducer<>(props);
           int userId = 1;

            while (userId < 1000000){
                A8 a = new A8(userId,"a1_"+userId,"a2_"+userId);
                B8 b = new B8(userId,"b1_"+userId,"b2_"+userId);
                C8 c = new C8(userId,"c1_"+userId,"c2_"+userId);
                D8 d = new D8(userId,"d1_"+userId,"d2_"+userId);
                E8 e = new E8(userId,"e1_"+userId,"e2_"+userId);
                F8 f = new F8(userId,"f1_"+userId,"f2_"+userId);
                G8 g = new G8(userId,"g1_"+userId,"g2_"+userId);
                H8 h = new H8(userId,"h1_"+userId,"h2_"+userId);
                I8 i = new I8(userId,"i1_"+userId,"i2_"+userId);
                J8 j = new J8(userId,"j1_"+userId,"j2_"+userId);
                ProducerRecord<String, A8> recordA = new ProducerRecord<>(KafkaTopic.ATopic8, String.valueOf(userId), a);
                ProducerRecord<String, B8> recordB = new ProducerRecord<>(KafkaTopic.BTopic8, String.valueOf(userId), b);
                ProducerRecord<String, C8> recordC = new ProducerRecord<>(KafkaTopic.CTopic8, String.valueOf(userId), c);
                ProducerRecord<String, D8> recordD = new ProducerRecord<>(KafkaTopic.DTopic8, String.valueOf(userId), d);
                ProducerRecord<String, E8> recordE = new ProducerRecord<>(KafkaTopic.ETopic8, String.valueOf(userId), e);
                ProducerRecord<String, F8> recordF = new ProducerRecord<>(KafkaTopic.FTopic8, String.valueOf(userId), f);
                ProducerRecord<String, G8> recordG = new ProducerRecord<>(KafkaTopic.GTopic8, String.valueOf(userId), g);
                ProducerRecord<String, H8> recordH = new ProducerRecord<>(KafkaTopic.HTopic8, String.valueOf(userId), h);
                ProducerRecord<String, I8> recordI = new ProducerRecord<>(KafkaTopic.ITopic8, String.valueOf(userId), i);
                ProducerRecord<String, J8> recordJ = new ProducerRecord<>(KafkaTopic.JTopic8, String.valueOf(userId), j);
                try {
                    producerA.send(recordA);
                    producerB.send(recordB);
                    producerC.send(recordC);
                    producerD.send(recordD);
                    producerE.send(recordE);
                    producerF.send(recordF);
                    producerG.send(recordG);
                    producerH.send(recordH);
                    producerI.send(recordI);
                    producerJ.send(recordJ);
                    System.out.println("Message sent successfully");
                    userId++;
                } catch (Exception exception) {
                    exception.printStackTrace();
                } finally {


                }
            }

        producerA.flush();
        producerB.flush();
        producerC.flush();
        producerD.flush();
        producerE.flush();
        producerF.flush();
        producerG.flush();
        producerH.flush();
        producerI.flush();
        producerJ.flush();

        producerA.close();
        producerB.close();
        producerC.close();
        producerD.close();
        producerE.close();
        producerF.close();
        producerG.close();
        producerH.close();
        producerI.close();
        producerJ.close();

    }

}